When a task instance is individually set to success or failure via the API, on_task_instance_success / on_task_instance_failed is called from the API server, giving listeners a terminal event to act on. The same does not happen when an entire Dag Run is force-set to success or failure: running tasks are bulk-transitioned in the database, but their task-instance listeners are never fired. Listeners that registered a start event (on_task_instance_running) for those tasks receive no completion event, leaving them — notably OpenLineage — with a job permanently stuck in RUNNING.

This PR closes the gap:

• _set_dag_run_terminal_state now returns (all_updated_tis, running_tis) where running_tis is the snapshot of task instances that were in an active state (RUNNING, DEFERRED, UP_FOR_RESCHEDULE, AWAITING_INPUT) before the bulk transition, captured without an extra query.
• set_dag_run_state_to_success / set_dag_run_state_to_failed expose the same tuple so callers have access to the running subset.
• patch_dag_run_state fires on_task_instance_success / on_task_instance_failed for those running task instances before the dag-run-level hook. Each group has its own try/except so a listener failure in one cannot suppress the other.
  Tasks that were only QUEUED or SCHEDULED (never started, no start event fired) are excluded — they transition to SKIPPED and receive no listener call, consistent with their lifecycle.

Why the terminal event is missing

The gap is a deliberate design decision that becomes a problem specifically for the dag run force-terminate case:

1. _set_dag_run_terminal_state writes SUCCESS/FAILED into the TI rows in the DB. No listener is fired.
2. Next heartbeat from the supervisor gets 409 CONFLICT ("TI is no longer running"). The supervisor sets self._terminal_state = SERVER_TERMINATED and kills the process via SIGTERM.
3. After the process dies, update_task_state_if_needed() runs. Because SERVER_TERMINATED ∈ STATES_SENT_DIRECTLY (supervisor.py:205), the condition at line 1461 is false — finish() is intentionally skipped. The reasoning is: "the server already knows the state." That is true — but the server setting the state and the server firing listeners for it are two different things, and the latter never happened.
4. Task runner finalizers never ran — the process was killed mid-execution before the on_task_instance_success/failed call in task-sdk/src/airflow/sdk/execution_time/task_runner.py:2249.
5. Scheduler's process_executor_events sees the executor report the task exit, but when it queries the TI it finds state=SUCCESS/FAILED (already written in step 1), so ti_queued is false (scheduler_job_runner.py:1460) and handle_failure is never called.

Result: on_task_instance_running fired from the task runner → task process killed → nothing on the other end. OpenLineage has a dangling RUNNING job with no terminal event. We can't extend the live of the worker process, but we can fire from the api server in that case.

Was generative AI tooling used to co-author this PR?

• Yes — Claude Code (claude-sonnet-4-6)

Generated-by: Claude Code (claude-sonnet-4-6) following the guidelines

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.